package com.tql.mall.rabbit.config;

import com.alibaba.fastjson.JSON;
import com.tql.mall.rabbit.constant.MqConstant;
import com.tql.mall.rabbit.model.MallCorrelationData;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;

/**
 * @Author: tyk
 * @Date: 2023/5/18 20:34
 * @Description:
 */
@Component
public class MqAckConfig implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnCallback {

    @Autowired
    private RabbitTemplate rabbitTemplate;
    @Autowired
    private RedisTemplate redisTemplate;

    /**
     * 指定ConfirmCallback和ReturnCallback
     * 只执行一次
     */
    @PostConstruct
    public void init() {
        rabbitTemplate.setConfirmCallback(this);            //指定 ConfirmCallback
        rabbitTemplate.setReturnCallback(this);             //指定 ReturnCallback
    }

    /**
     * 交换机确认回调方法：确认消息是否成功到达交换机
     *
     * @param data  消息id及相关信息
     * @param ack   对生产者的应答（true表示交换机成功收到消息）
     * @param cause 未到达的原因
     */
    @Override
    public void confirm(CorrelationData data, boolean ack, String cause) {
        // 消息没有成功到达交换机则重新发送消息
        if (!ack) this.retry(data);
    }


    /**
     * 队列确认回调方法：确认消息是否成功到达队列（仅当消息未到达队列时执行）
     *
     * @param message    消息主体
     * @param replyCode  应答码
     * @param replyText  描述
     * @param exchange   交换器
     * @param routingKey 路由
     */
    @Override
    public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
        // 获取消息ID
        String id = (String) message.getMessageProperties().getHeaders().get(MqConstant.SPRING_RETURNED_MESSAGE_CORRELATION);
        // 获取持久化的消息（从redis中获取）
        String strJson = (String) redisTemplate.opsForValue().get(id);
        MallCorrelationData data = JSON.parseObject(strJson, MallCorrelationData.class);
        // 重新发送消息
        this.retry(data);
    }

    /**
     * 重新发送消息
     */
    private void retry(CorrelationData correlationData) {
        // 转换
        MallCorrelationData data = (MallCorrelationData) correlationData;
        // 获取可重试次数
        int retryCount = data.getRetryCount();
        // 可重试次数大于0则重试
        if (retryCount > 0) {
            // 可重试次数-1
            data.setRetryCount(--retryCount);
            // 更新Redis
            redisTemplate.opsForValue().set(data.getId(), JSON.toJSONString(data));
            // 延迟消息
            if (data.isDelay()) {
                rabbitTemplate.convertAndSend(
                        data.getExchange(),
                        data.getRoutingKey(),
                        data.getMsg(),
                        message -> {
                            message.getMessageProperties().setDelay(data.getDelayTime());
                            return message;
                        },
                        data);
            }
            // 普通消息
            else rabbitTemplate.convertAndSend(data.getExchange(), data.getRoutingKey(), data.getMsg(), data);

        }
    }
}

